package com.uptown.matrix.core.producer.instance;

import java.util.Map;

import javax.annotation.Resource;

import com.uptown.matrix.common.result.WebResult;
import com.uptown.matrix.core.producer.ProducerService;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.stereotype.Service;

@Service
public class RedisProducerService extends ProducerService {

    @Resource
    RedisTemplate stringRedisTemplate;

    @Resource
    Map<String, StreamMessageListenerContainer<String, ObjectRecord<String, Object>>> redisConsumerMap;


    @Override
    public WebResult producerSend(String bootServer, String group, String topic, String message) {
        // FIXME: 白名单校验
        super.producerSend(bootServer, group, topic, message);

        ObjectRecord<String, Object> stringUserObjectRecord = ObjectRecord.create(topic, message);
        RecordId recordId = stringRedisTemplate.opsForStream().add(stringUserObjectRecord);

        // value为ack
        return WebResult.of(recordId);
    }

    /**
     * topic下历史所有数据
     */
    public WebResult metaDataList(String topic) {

        return WebResult.of(stringRedisTemplate.opsForStream().range(topic,
                Range.of(Range.Bound.inclusive("-"),
                        Range.Bound.inclusive("+"))));
    }

    /**
     * topic下未提交的数据
     */
    public WebResult ackMetaDataList(String topic, String group) {
        return WebResult.of(stringRedisTemplate.opsForStream().pending(topic, group));
    }

    /**
     * 区间数据
     */
    public WebResult rangeFrom(String topic, String from, String to) {
        return WebResult.of(stringRedisTemplate.opsForStream().range(topic,
                Range.of(Range.Bound.inclusive(from), Range.Bound.inclusive(to))));
    }

    @Override
    public WebResult topicList() {
        // 白名单
        return WebResult.of(redisConsumerMap.keySet());
    }
}
